package org.example.rocketmq.msg_type.order;

import org.apache.commons.lang3.ThreadUtils;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.example.rocketmq.NamesrvAddr;

import java.io.UnsupportedEncodingException;
import java.time.Duration;
import java.util.List;

/**
 * 顺序消息的生产者
 */
public class ProducerOrder {

    public static void main(String[] args) throws Exception{
        producer();
    }


    /**
     * 顺序消息值生产者局部有序发送到一个 queue, 但多个 queue 之间是全局无序的。
     * 顺序消息生产者样例：通过 MessageQueueSelector将消息有序发送到同一个queue中
     * @throws Exception
     */
    public static void producer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("producer_order_group_01");
        producer.setNamesrvAddr(NamesrvAddr.get());
        producer.start();

        for (int i = 0; i < 5; i++) {
            for (int j = 0; j < 10; j++) {
                Message message = new Message("TopicTest", "TagA", ("Hello RocketMQ Order" + i + "_step_" + j).getBytes(RemotingHelper.DEFAULT_CHARSET));

                SendResult result = producer.send(message, new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                        Integer id = (Integer) o;
                        int index = id % list.size();
                        System.out.println("消息队列的个数：" + list.size() + ", 选择的消息队列是：" + index);
                        return list.get(index);
                    }
                }, i);

                System.out.printf("消息发送成功 %s%n", result);
                ThreadUtils.sleep(Duration.ofSeconds(1));
            }

        }

        producer.shutdown();
    }

}
